package com.bw.flinkstreaming.state1.job5;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class ReducingStateWithCountAvg extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Long>> {

    //定义状态
    private ReducingState<Long> sumState;

    @Override
    public void open(Configuration parameters) throws Exception {
        //创建
        ReducingStateDescriptor reducingStateDescriptor = new ReducingStateDescriptor<Long>(
                "avg",

                /**
                 * 聚合函数
                 * Long value1：累加的值
                 * Long value2：当前获取的数据的值
                 */
                new ReduceFunction<Long>() {
                    @Override
                    public Long reduce(Long value1, Long value2) throws Exception {
                        System.out.println("value1: "+value1);
                        System.out.println("value2: "+value2);
                        return value1+value2;
                    }
                },Long.class);
        sumState = getRuntimeContext().getReducingState(reducingStateDescriptor);
    }

    @Override
    public void flatMap(Tuple2<Long, Long> element, Collector<Tuple2<Long, Long>> out) throws Exception {
        sumState.add(element.f1);
        out.collect(Tuple2.of(element.f0,sumState.get()));
    }
}
